package com.heima.redisson.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class DelayTaskListener {

    private ThreadPoolExecutor taskMsgThreadPool = new ThreadPoolExecutor(1, 1,
            120, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

    private ThreadPoolExecutor bizThreadPool  = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors()*2,
            120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(20));

    @Autowired
    private RedissonClient redissonClient;

    /**
     * bean的生命周期回调
     * */
    @PostConstruct
    public void postConstruct(){
        // 子线程，while(true)不停的从延迟队列中取消息
        taskMsgThreadPool.submit(new Runnable() {
            @Override
            public void run() {
                while(true){
                    // 从延迟队列中取消息
                    RBlockingDeque<String> demoQueue = redissonClient.getBlockingDeque("demo_queue");
                    try{
                        String value = demoQueue.take();
                        bizThreadPool.submit(new Runnable() {
                            @Override
                            public void run() {
                                // 做业务逻辑
                                checkOrderStatus();
                            }
                        });
                        log.info("延迟任务出队：{}", value);
                    }catch(Exception e){
                        log.error("从阻塞队列获取消息异常:{}", ExceptionUtils.getStackTrace(e));
                    }
                }
            }
        });
    }
    // 30分支不知福 自动取消
    // 会议开始前30分钟，发消息提醒
    // 10:00 用户下了一个订单，向redisson延迟队列发一个延迟30分钟的消息，10:30 消息要出队
    // 11:00 要去开会，10:30需要给提醒

    private void checkOrderStatus(){
        try {
            // 业务逻辑执行了30分钟
            Thread.sleep(30 * 60 * 100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}
